package com.itmake.sample.mqtt.consumer;

import com.itmake.sample.mqtt.config.MqttConfig;
import com.itmake.sample.mqtt.constant.MqttConstant;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

/**
 * MQTT消费者配置
 *
 * @author Amjacks
 * @since 2023/08/10
 */
@Configuration
@IntegrationComponentScan
public class MqttConsumerConfig {

	@Resource
	private MqttConfig mqttConfig;

	@Resource
	private MqttPahoClientFactory mqttClientFactory;

	@Resource
    private MqttConsumer mqttConsumer;

    /**
     * 信息通道
     *
     * @return {@link MessageChannel}
     */
    @Bean(value = MqttConstant.IN_BOUND_CHANNEL)
    public MessageChannel mqttInBoundChannel() {
        return new DirectChannel();
    }

    /**
     * 消息订阅绑定
     */
    @Bean
    public MessageProducer inbound() {
        // 所有的 topic
        List<String> strings = new ArrayList<>(mqttConfig.getConsumer().values());
        // 不存在需要监听的 topic 直接返回 null
        if (CollectionUtils.isEmpty(strings)) {
            return null;
        }

        // 创建 adapter 监听 topic
        String[] res = new String[strings.size()];
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
                        mqttConfig.getClientInId(), mqttClientFactory, strings.toArray(res));
        adapter.setCompletionTimeout(mqttConfig.getTimeout());
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(mqttConfig.getQos());
        adapter.setOutputChannel(mqttInBoundChannel());
        return adapter;
    }


    /**
     * 消息处理器
     *
     * @return {@link MessageHandler}
     */
    @Bean
    @ServiceActivator(inputChannel = MqttConstant.IN_BOUND_CHANNEL)
    public MessageHandler handler() {
        return message -> {
            // 消息接收到之后，通过消息分发组件进行处理
            String topic = String.valueOf(message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC));
            String msg = message.getPayload().toString();
            // 处理消息
            mqttConsumer.handlerMsg(topic, msg);
        };
    }
}
